package kafka;

import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.PartitionInfo;

import java.util.*;

import static java.lang.Boolean.FALSE;

/**
 * USER: lintc 【lintiancong@zhuojianchina.com】
 * DATE: 2017-02-06 17:28
 */
public class KafkaUtil {

    /**
     * 通过线程MAP管理 并新增线程 （无法删除线程）
     * 1 获取topic列表
     * 2 创建相应的线程抓取日志
     * 3 定时查询 topic 列表 并查找相应线程
     * 4 删除不存在的 topic线程 新建新的topic线程
     * @param args
     * @throws InterruptedException
     */
    public static void main(String[] args) throws InterruptedException {
//        List<String> list;
//        Map<String, Thread> customerList = new HashMap<String, Thread>();
//        Thread thread;
//        while (true) {
//            list = getTopicList(KafkaCustomer.getConsumer());
//            for (int i = 0; i < list.size(); i++) {
//                final String topicName = list.get(i);
//                System.out.println(topicName);
//                if(null == customerList.get(topicName)) { // 判断该 topic 是否已存在
//                    thread = new Thread(new Runnable() {
//                        public void run() {
//                            KafkaCustomer.getMsg(topicName);
//                        }
//                    });
//                    thread.setName(topicName);
//                    thread.start();
//                    customerList.put(topicName, thread);
//                } else {
//                    System.out.println(topicName + " : 已存在");
//                }
//            }
//            Thread.sleep(5 * 60 * 1000);
//            //  Thread.sleep(20 * 1000);
//        }
//        getTopicList(KafkaCustomer.getConsumer());
        KafkaConsumer<String, String> consumer = KafkaCustomer.getConsumer();
        consumer.close();
        consumer.subscribe(Arrays.asList("app_zyyy"));
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(50);
        }
//        consumer.
    }

    /**
     * 获取 topic 列表
     * @param consumer
     */
    public static List<String> getTopicList(KafkaConsumer<String, String> consumer) {
        Map<String, List<PartitionInfo>> map =  consumer.listTopics();
        List<String> list = new ArrayList<String>();
        for (String s : map.keySet()) {
            System.out.println(s);
            list.add(s);
        }
        return list;
    }

}
